package study.rocketmq.usespringboot;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import study.rocketmq.vo.Order;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

@Service
public class SprintBootProducerService {
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer messageTimeOut;
    public  void send(Order order){
        rocketMQTemplate.convertAndSend("springboot-topic-order",order);
    }

    /**
     * 发送延时消息<br/>
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h<br/>
     */
    public SendResult sendDelayMsg(String msgBody, Integer delayLevel){
        return rocketMQTemplate.syncSend("springboot-topic-delay", MessageBuilder.withPayload(msgBody).build(),messageTimeOut,delayLevel);
    }

    /**
     * 发送带tag的消息,直接在topic后面加上":tag"
     */
    public SendResult sendTagMsg(String msgBody){
        return rocketMQTemplate.syncSend("springboot-topic-tag:tag1",MessageBuilder.withPayload(msgBody).build());
    }

    /**
     * 顺序消息 - 单个发送
     */
    public void sendOrderlyMsg(  ){

            // send to 4 queues
            for (int i = 0; i < 10; i++) {
                Order order = new Order(i+"","msg_"+i);
                if(i % 2 == 0) {
                    rocketMQTemplate.syncSendOrderly("springboot-topic-orderly", MessageBuilder.withPayload(order).build(), "0", 60000);
                }else {
                    rocketMQTemplate.syncSendOrderly("springboot-topic-orderly", MessageBuilder.withPayload(order).build(),  "1", 60000);
                }
            }
            // 批量发送
            List<Message> msgs = new ArrayList<Message>();
            for (int i = 0; i < 10; i++) {
                Order order = new Order("batch_id_"+i,"msg_"+i);
                msgs.add(MessageBuilder.withPayload(order).build());
            }
            rocketMQTemplate.syncSendOrderly("springboot-topic-orderly", msgs,  "1", 60000);

        }
    /**
     * 顺序消息 - 批量发送
     */
    public void batchSendOrderlyMsg(){

        for (int q = 0; q < 4; q++) {
            // send to 4 queues
            List<Message> msgs = new ArrayList<Message>();
            for (int i = 0; i < 10; i++) {
                int msgIndex = q * 10 + i;
                String msg = String.format("Hello RocketMQ Batch Msg#%d to queue: %d", msgIndex, q);
                msgs.add(MessageBuilder.withPayload(msg).
                        setHeader(RocketMQHeaders.KEYS, "KEY_" + msgIndex).build());
            }
            SendResult sr = rocketMQTemplate.syncSendOrderly("springboot-topic-orderly", msgs, q + "", 60000);
            System.out.println("--- Batch messages orderly to queue :" + sr.getMessageQueue().getQueueId() + " send result :" + sr);
        }

    }

}

